package com.atguigu.app.dwm;

import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;

public class UniqueVisitApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //设置CK & 状态后端
        //env.enableCheckpointing(5000L);
        //env.getCheckpointConfig().setCheckpointTimeout(5000L);
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink-210108/cdc/ck"));

        //TODO 2.读取Kafka数据创建流并转换为JSONObject
        String groupId = "unique_visit_app_210108";
        String sourceTopic = "dwd_page_log";
        String sinkTopic = "dwm_unique_visit";
        SingleOutputStreamOperator<JSONObject> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId))
                .map(JSONObject::parseObject);

        //TODO 3.按照Mid分组
        KeyedStream<JSONObject, String> keyedStream = kafkaDS
                .keyBy(data -> data.getJSONObject("common").getString("mid"));

        //TODO 4.过滤
        SingleOutputStreamOperator<JSONObject> filterDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {

            //声明状态 访问时间
            private ValueState<String> valueState;

            //时间格式化对象
            private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stringValueStateDescriptor = new ValueStateDescriptor<>("last-visit", String.class);

                //为了在状态无效时清理,放置状态数据量过大
                StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build();

                stringValueStateDescriptor.enableTimeToLive(ttlConfig);
                valueState = getRuntimeContext().getState(stringValueStateDescriptor);
                sdf = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {

                //获取上一跳页面信息
                String lastPage = value.getJSONObject("page").getString("last_page_id");

                //判断是否存在上一跳
                if (lastPage == null || lastPage.length() <= 0) {

                    //疑似目标数据,获取状态数据
                    String lastDate = valueState.value();
                    String date = sdf.format(value.getLong("ts"));

                    if (lastDate == null || !lastDate.equals(date)) {

                        //将今天的时间写入状态
                        valueState.update(date);

                        //保留该数据
                        return true;
                    } else {
                        return false;
                    }

                } else {
                    //过滤掉
                    return false;
                }
            }
        });

        //TODO 5.写入Kafka主题
        filterDS.print(">>>>>>>>>");
        filterDS.map(data -> JSONObject.toJSONString(data));
        filterDS.map(JSONAware::toJSONString)
                .addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        //TODO 6.启动
        env.execute();

    }

}
